package com.javafactroy.kunlong.bip.icp.message.test.netty;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultPromise;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.javafactroy.kunlong.bip.icp.common.config.BipConfig;
import com.javafactroy.kunlong.bip.icp.common.consts.EAppType;
import com.javafactroy.kunlong.bip.icp.context.AppContext;
import com.javafactroy.kunlong.bip.icp.message.base.iface.IMessage;
import com.javafactroy.kunlong.bip.icp.message.test.base.TMsg;
import com.javafactroy.kunlong.bip.icp.message.test.netty.handler.client.TMsgClientHandlerInitilizer;
import com.javafactroy.kunlong.bip.icp.message.test.netty.handler.client.TMsgSyncFuture;
import com.javafactroy.kunlong.bip.icp.message.test.netty.handler.client.TMsgSyncFutureMngr;

import java.util.Properties;
import java.util.concurrent.TimeUnit;


/**
 * @Description: Netty客户端
 * @Author:leijmdas
 * @Since:2019年9月26日下午8:54:59
 */
@Component
@Data
@Slf4j
public class NettyClientTMsg {

    @Autowired
    BipConfig bipConfig;
    EAppType appType;

    public NettyClientTMsg() {
    }

    private static volatile EventLoopGroup group;// = new NioEventLoopGroup();

    public void close() throws InterruptedException {
        getSocketChannel().close().sync();

    }

    /**
     * @Fields socketChannel : 通道
     */
    private SocketChannel socketChannel;

    /**
     * @Fields clientHandlerInitilizer : 初始化
     */

    private TMsgClientHandlerInitilizer initilizer;

    /**
     * @Description: 启动客户端
     * @Author: 雷建明
     * @Since: 2023:43:21
     */
    @SuppressWarnings("unchecked")
    public synchronized void start(EAppType appType) throws InterruptedException {

        this.appType = appType;
        log.info("****appType = {}", appType);

        initilizer = AppContext.getBean(TMsgClientHandlerInitilizer.class);
        if (group == null) {
            synchronized (NettyClientTMsg.class) {
                if (group == null) {
                    group = new NioEventLoopGroup();
                }
            }
        }

        String hostIp = bipConfig.getServerIp();
        int port = bipConfig.getServerPort();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                // 指定Channel
                .channel(NioSocketChannel.class)
                .remoteAddress(hostIp, port)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(initilizer);

        // 连接
        ChannelFuture channelFuture = bootstrap.connect().sync();
        socketChannel = (SocketChannel) channelFuture.channel();
        //客户端断线重连逻辑
        channelFuture.addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("连接Netty服务端成功...");
                } else {

                    log.info("连接Netty服务端失败，进行断线重连...");
                    final EventLoop loop = future.channel().eventLoop();
                    loop.schedule(new Runnable() {
                        @SneakyThrows
                        @Override
                        public void run() {
                            log.info("连接正在重试...");
                            //start(appType);
                        }
                    }, 30, TimeUnit.SECONDS);
                }
            }
        });


    }


    public IMessage sendSyncMsg(IMessage message, TMsgSyncFuture<IMessage> syncFuture) {

        TMsg msg = (TMsg) message;
        try {

            ChannelFuture future = socketChannel.writeAndFlush(msg);
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {

                    if (future.isSuccess()) {
                        log.info("===========发送成功！");
                    } else {
                        log.error("-----------发送失败！！！");
                        TMsgSyncFutureMngr.ackSyncMsgErr(message);
                    }
                }
            });
            return syncFuture.get(15, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getCause());
        }


    }

    public void setIpPort(String ip, int port) {
        if (bipConfig == null) {
            bipConfig = new BipConfig();
            bipConfig.setProps(new Properties());
        }
        if (bipConfig.getProps() == null) {
            bipConfig.setProps(new Properties());
        }
        bipConfig.getProps().setProperty("server.ip", ip);
        bipConfig.getProps().setProperty("server.port", String.valueOf(port));
    }

    public IMessage send(IMessage message) {
        return sendSyncMsg(message);
    }

    public IMessage sendSyncMsg(IMessage message) {

        TMsgSyncFuture<IMessage> syncFuture = new TMsgSyncFuture<>();

        TMsgSyncFutureMngr.put(message.getMessageId(), syncFuture);
        return sendSyncMsg(message, syncFuture);


    }

    /**
     * sendMsg by promise
     */
    public TMsg sendTMsg(TMsg msg) {
        return (TMsg) sendPromise(msg);
    }

    public IMessage sendPromise(IMessage message) {
        DefaultPromise<IMessage> promise = new DefaultPromise<IMessage>(socketChannel.eventLoop());//

        try {
            initilizer.setPromise(promise);
            ChannelFuture future = socketChannel.writeAndFlush(message);
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {

                    if (future.isSuccess()) {
                        log.info("===========发送成功！");
                    } else {
                        log.error("-----------发送失败！！！");
                        promise.setFailure(promise.cause());
                        initilizer.setPromise(null);
                     }
                }
            });
            promise.await();
            if (promise.isSuccess()) {
                return promise.getNow();//(10, TimeUnit.SECONDS);
            } else {
                throw new RuntimeException(promise.cause());
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e.getCause());
        }

    }

}
